package com.train.mq.support;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Set;

@Slf4j
@Component
public class RocketMQSupport {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 发送同步消息携带自定义请求头
     * @param topic
     * @param eventBody
     * @param eventHeader
     * @param <T1>
     * @param <T2>
     */
    public <T1, T2> void sendSyncMessage(String topic, T1 eventBody, T2 eventHeader) {
        Map<String, Object> eventHeaders = JSON.parseObject(JSON.toJSONString(eventHeader), Map.class);
        sendSyncMessage(topic, eventBody, eventHeaders);
    }

    /**
     * 发送同步消息
     * @param topic
     * @param eventBody
     * @param <T>
     */
    public <T> void sendSyncMessage(String topic, T eventBody) {
        sendSyncMessage(topic, eventBody, null);
    }

    /**
     * 发送同步消息
     * @param topic
     * @param eventBody
     * @param eventHeaders
     * @param <T>
     */
    public <T> void sendSyncMessage(String topic, T eventBody, Map<String, Object> eventHeaders) {
        MessageBuilder<String> builder = MessageBuilder.withPayload(JSON.toJSONString(eventBody));
        if (MapUtils.isNotEmpty(eventHeaders)) {
            Set<String> keySet = eventHeaders.keySet();
            for (String key : keySet) {
                builder.setHeader(key, eventHeaders.get(key));
            }
        }
        SendResult sendResult = rocketMQTemplate.syncSend(topic, builder.build());
        log.info("send message result status: {}", sendResult.getSendStatus());
    }
}
